package com.mjf.chain

import org.apache.flink.streaming.api.scala._

object StartNewChainDemo {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val source: DataStream[Int] = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

    val result: DataStream[String] = source
      .filter(_.>(5))
      .startNewChain()  // 开启一个新的任务链，前面的算子不能合并到同一个slot
      .map(_ + "大于5")

    result.print()

    env.execute(StartNewChainDemo.getClass.getName)

  }
}
